package com.example.spark.rdd;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.AbstractJavaRDDLike;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;

public class RDDWordCount_3 {

    public static void main(String[] args) {
        //创建配置对象
        SparkConf conf = new SparkConf();
        conf.setMaster("local[*]");
        conf.setAppName("WordCount");

        //创建Spark的上下文
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaPairRDD<String, Integer> wcRDD = jsc.textFile("input/2.txt")
                .flatMap(line -> Arrays.asList(line.split(" ")).iterator())
                .mapToPair(word -> new Tuple2<>(word, 1))
                .reduceByKey(((v1, v2) -> v1 + v2));
        wcRDD.collect().forEach(System.out::println);
        jsc.stop();

//        //编写逻辑代码
//        //读取input目录下的words.txt文件的数据
//        JavaRDD<String> rdd = jsc.textFile("input/2.txt");
//        //使用FlatMap转换功能把长字符串切分单个单词
//        JavaRDD<String> flatMapRDD=rdd.flatMap(line->Arrays.asList(line.split(" ")).iterator());
///*        JavaRDD<String> flatMapRDD = rdd.flatMap(new FlatMapFunction<String, String>() {
//            @Override
//            public Iterator<String> call(String line) throws Exception {
//                String[] words = line.split(" ");
//                List<String> list = Arrays.asList(words);
//
//                return list.iterator();
//            }
//        });
//
// */
//
//        //输出测试
//       // print(flatMapRDD);
//
//        //转换格式（word, 1）
//        JavaPairRDD<String, Integer> pairRDD=flatMapRDD.mapToPair(word->new Tuple2<>(word,1));
///*      JavaPairRDD<String, Integer> pairRDD = flatMapRDD.mapToPair(new PairFunction<String, String, Integer>() {
//             @Override
//            public Tuple2<String, Integer> call(String word) throws Exception {
//                 //Tuple2：[(单词,次数)，(单词,次数)，(单词,次数)]
//                 return new Tuple2<>(word,1);
//            }
//         });
//
// */
//        //输出测试
//       // print(pairRDD);
//        //合并相同的单词 (hello,1),(hello,1) => (hello,2)
//        JavaPairRDD<String, Integer>  reduceByKeyRDD=pairRDD.reduceByKey(((v1, v2) -> v1+v2));
//        /*
//        JavaPairRDD<String, Integer> reduceByKeyRDD = pairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
//
//            @Override
//            public Integer call(Integer v1, Integer v2) throws Exception {
//
//                //(hello,1),(hello,1) => (hello,2) 即：1+1=2
//                return v1+v2;
//            }
//        });
//        */
//
//        //输出
//        print(reduceByKeyRDD);
//
//        //回收资源
//        jsc.stop();


    }

    public static void print(AbstractJavaRDDLike rdd){
        rdd.collect().forEach(System.out::println);

    }
}
